package com.sxt.disruptor.server;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.springframework.stereotype.Component;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sxt.disruptor.quickstart.StringEvent;

/**
 * DisruptorServer，封装了Disruptor服务，提供 start ，stop，注册消费者，提供发布方法等功能
 * 
 * 该类 可以直接使用，并且独立于业务。该类描述了 Disruptor的核心
 * 
 * 
 * @author Administrator
 * @version v1.0.0
 */
@Component
public class DisruptorServer1 {
	//环形数组
	private RingBuffer<Trade> ringBuffer = null;
	//生产者
	private TradeProducer tradeProducer = null;
	
	// 静态内部类的单例模式
		private static class SingletonHolder {
			private static final DisruptorServer1 INSTANCE = new DisruptorServer1();
		}

		// 对外创建
		public static DisruptorServer1 getInstance() {
			return SingletonHolder.INSTANCE;
		}

		// 对外不能暴露的接口
		private DisruptorServer1() {
		}
		
		public void start() {
			initDisruptor();
		}
	
	/**
	 * 	获取一个生产者
	 * @return TradeProducer
	 */
	public TradeProducer getProducer() {
		return this.tradeProducer;
	}
	/**
	 * 
	 * @return
	 */
	public RingBuffer<Trade> getRingBuffer(){
		return this.ringBuffer;
	}
	
	
	private static final int BUFFER_SIZE = 1024 * 1024;// 环形队列长度，必须是2的N次方
	
	private Disruptor<Trade> disruptor;

	

	@SuppressWarnings("deprecation")
	private void initDisruptor() {
		// 1.创建一个线程工厂提供线程来触发Consumer的事件处理
		ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()/2);// 线程池，构建disruptor需要
		//ThreadFactory threadFactory = Executors.defaultThreadFactory();
		// 2.创建工厂
		EventFactory<Trade> factory = new EventFactory<Trade>() {
			@Override
			public Trade newInstance() {
				return new Trade();
			}
		};
		
		//3.创建ringBuffer容器
        ringBuffer = RingBuffer.create(ProducerType.MULTI, factory, BUFFER_SIZE, new BlockingWaitStrategy());
		//4.设置序号栅栏
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //创建消费者数组
        TradeHandler[] consumers = new TradeHandler[5];
        
        for(int i = 0; i<consumers.length;i++) {
        	consumers[i] = new TradeHandler();
        }
        //5.设置工作池
        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(
                this.ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(), consumers);
        //5 添加我们的sequences
        this.ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
         
        //6 启动我们的工作池
        workerPool.start(executor);
        
        //
        this.tradeProducer = new TradeProducer(ringBuffer);
	}
	
	public static void main(String[] args) {
		DisruptorServer1.getInstance().start(); //初始化服务 
		
		TradeProducer tradeProducer = DisruptorServer1.getInstance().getProducer();
		
		
		String data = "tradeProducer1111111111";
		
		tradeProducer.onData(data);
	}
}
